package patterns

import (
	"fmt"
	"math/rand"
)

// 无并发版本
func TryPipelines() {
	multiply := func(values []int, multiplier int) []int {
		multiplierValues := make([]int, len(values))
		for i, v := range values {
			multiplierValues[i] = v * multiplier
		}
		return multiplierValues
	}

	add := func(values []int, additive int) []int {
		addValues := make([]int, len(values))
		for i, v := range values {
			addValues[i] = v + additive
		}
		return addValues
	}
	ints := []int{1, 2, 3, 4}
	for _, v := range add(multiply(ints, 2), 1) {
		println(v)
	}
}

func TryPipelines2() {
	generate := func(done <-chan interface{}, nums []int) <-chan int {
		intStream := make(chan int)

		go func() {
			defer close(intStream)
			for _, n := range nums {
				select {
				case <-done:
					return
				case intStream <- n:
				}
			}
		}()
		return intStream
	}
	multiply := func(done <-chan interface{}, intStream <-chan int, multiplier int) <-chan int {
		multiplierStream := make(chan int)
		go func() {
			defer close(multiplierStream)
			for num := range intStream {
				select {
				case <-done:
					return
				case multiplierStream <- num * multiplier:
				}
			}
		}()
		return multiplierStream
	}
	add := func(done <-chan interface{}, intStream <-chan int, addier int) <-chan int {
		addStream := make(chan int)
		go func() {
			defer close(addStream)
			for num := range intStream {
				select {
				case <-done:
					return
				case addStream <- num + addier:
				}
			}
		}()
		return addStream
	}
	done := make(chan interface{})
	defer close(done)
	pipline := multiply(done, add(done, multiply(done, generate(done, []int{1, 2, 3, 4}), 2), 1), 2)
	for v := range pipline {
		fmt.Println(v)
	}

}

func TryPipelinesRepeat() {
	var toString = func(done <-chan interface{}, valueStream <-chan interface{}) <-chan string {
		stringStream := make(chan string)

		go func() {
			defer close(stringStream)

			for v := range valueStream {
				select {
				case <-done:
					return
				case stringStream <- v.(string):
				}
			}
		}()

		return stringStream
	}

	var repeat = func(done <-chan interface{}, values ...interface{}) <-chan interface{} {
		valueStream := make(chan interface{})

		go func() {
			defer close(valueStream)

			for {
				for _, v := range values {
					select {
					case <-done:
						return
					case valueStream <- v:
					}
				}
			}
		}()

		return valueStream
	}

	var take = func(done <-chan interface{}, valueStream <-chan interface{}, num int) <-chan interface{} {
		takeStream := make(chan interface{})

		go func() {
			defer close(takeStream)

			for i := 0; i < num; i++ {
				select {
				case <-done:
					return
				case takeStream <- <-valueStream:
				}
			}
		}()

		return takeStream
	}

	var repeatFn = func(done <-chan interface{}, fn func() interface{}) <-chan interface{} {
		valueStream := make(chan interface{})

		go func() {
			defer close(valueStream)

			for {
				select {
				case <-done:
					return
				case valueStream <- fn():
				}
			}
		}()

		return valueStream
	}
	done := make(chan interface{})
	defer close(done)

	for num := range take(done, repeat(done, 1), 10) {
		fmt.Printf("%v ", num)
	}

	rand := func() interface{} {
		return rand.Int()
	}

	for num := range take(done, repeatFn(done, rand), 10) {
		fmt.Println(num)
	}

	var message string
	for token := range toString(done, take(done, repeat(done,
		"I", "am."), 3)) {
		message += token
	}

	fmt.Println("message:", message)
}
